package com.jubull.rule.engine.sources;

import com.jubull.rule.engine.utils.ConfigNames;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * @author liuyun
 * @title: KafkaSourceBuilder
 * @projectName jubull_marking
 * @description: 各类kafka的SourceBuilder
 * @date 2022-3-25 00:53
 */
public class KafkaSourceBuilder {

    private final Config config;

    public KafkaSourceBuilder() {
        config = ConfigFactory.load();
    }

    public FlinkKafkaConsumer<String> build(String topic) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", config.getString(ConfigNames.KAFKA_BOOTSTRAP_SERVER));
        props.setProperty("auto.offset.reset", config.getString(ConfigNames.KAFKA_AUTO_OFFSET_RESET));
        return new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), props);
    }
}
